[新機能]Amazon Redshift Spectrum がついにCTASとINSERT INTOをサポートしたので実際に試してみました！

データアナリティクス事業本部の石川です。クラスタバージョン1.0.15582から 「クエリの結果から外部テーブルを作成する（CTAS）」と「追加するテーブルの作成」が、Redshift Spectrumでも利用できるようになりました。早速試してみます！

CTASに関してはこれまでローカルテーブルへは出来ましたが外部テーブルへのCTASでの書き込みができるようになったのがポイントです。今まで同じことをしようとUnload+Create external tableと2ステップで行う必要がありました。

— Junpei Ozono (@jostandard) June 8, 2020

準備

検証用にサンプルデータのCSVファイルを用いて、外部スキーマと外部テーブルを作成します。

外部スキーマの作成

cmdb=# create external schema cm_user_db
cmdb-# from data catalog
cmdb-# database 'cm_user_db'
cmdb-# iam_role 'arn:aws:iam::1234567890123:role/redshift-role'
cmdb-# create external database if not exists;
INFO:  External database "cm_user_db" already exists
CREATE SCHEMA

外部テーブルの作成

cmdb=# create external table cm_user_db.order(
cmdb(# row_id bigint,
cmdb(# order_id varchar(32),
cmdb(# order_date varchar(10),
cmdb(# ship_date varchar(10),
cmdb(# ship_mode varchar(64),
cmdb(# customer_id varchar(64),
cmdb(# customer_name varchar(64),
cmdb(# segment varchar(64),
cmdb(# country varchar(16),
cmdb(# city varchar(16),
cmdb(# state varchar(16),
cmdb(# region varchar(16),
cmdb(# product_id varchar(255),
cmdb(# category varchar(16),
cmdb(# sub_category varchar(32),
cmdb(# product_name varchar(255),
cmdb(# sales double precision,
cmdb(# quantity bigint,
cmdb(# discount double precision,
cmdb(# profit double precision)
cmdb-# row format delimited
cmdb-# fields terminated by '\t'
cmdb-# stored as textfile
cmdb-# location 's3://cm-user/orders_unload/'
cmdb-# table properties ('numRows'='15889', 'skip.header.line.count'='1');
CREATE EXTERNAL TABLE

データの確認

cmdb=# select * from cm_user_db.order limit 1;
-[ RECORD 1 ]-+----------------------------
row_id        | 3560
order_id      | JP-2014-2034412
order_date    | 2014-01-06
ship_date     | 2014-01-08
ship_mode     | セカンド クラス
customer_id   | 桃優-13840
customer_name | 桃木 優太
segment       | 大企業
country       | 日本
city          | 福山
state         | 広島
region        | 中国地方
product_id    | 家具-本棚-10003403
category      | 家具
sub_category  | 本棚
product_name  | Dania コーナー シェルフ, 白
sales         | 24918
quantity      | 3
discount      | 0
profit        | 2490

クエリの結果から外部テーブルを作成する（CTAS）

構文

CREATE EXTERNAL TABLE
external_schema.table_name  
[ PARTITIONED BY (col_name [, … ] ) ] 
[ ROW FORMAT DELIMITED row_format ]
STORED AS file_format
LOCATION { 's3://bucket/folder/' }
[ TABLE PROPERTIES ( 'property_name'='property_value' [, ...] ) ]
 AS
 { select_statement }

外部テーブルの作成＆データの追加

orderから顧客毎の売上と利益のデータマートを作成しています。データマートとなる外部テーブルのデータはs3://cm-user/order_summary/に保存します。

cmdb=# create external table cm_user_db.order_summary
cmdb-# stored as parquet
cmdb-# location 's3://cm-user/order_summary/' as (
cmdb(# select
cmdb(# customer_id,
cmdb(# customer_name,
cmdb(# sum(sales) as total_sales,
cmdb(# sum(profit) as total_profit
cmdb(# from cm_user_db.order
cmdb(# group by 1,2
cmdb(# );
INFO:  794 record(s) exported successfully.
CREATE EXTERNAL TABLE

結果の確認

以下の通り、上記のクエリを実行しただけで、外部テーブルが作成できました。

cmawsteamdb=# select * from cm_user_db.order_summary limit 1;
-[ RECORD 1 ]-+-----------
customer_id   | 岩涼-13855
customer_name | 岩名 涼
total_sales   | 262146.8
total_profit  | 23871.8

指定したS3のパスには４つに分割されたParquetファイルが出力されました。恐らく、検証環境のクラスタのスライス数と同数のファイルに分割されたと考えられます。

% aws s3 ls s3://cm-user/order_summary/
2020-06-09 08:29:49       9573 20200608_232943_316378_1485945_0000_part_00.parquet
2020-06-09 08:29:49       8603 20200608_232943_316378_1485945_0001_part_00.parquet
2020-06-09 08:29:49       9171 20200608_232943_316378_1485945_0002_part_00.parquet
2020-06-09 08:29:49       8199 20200608_232943_316378_1485945_0003_part_00.parquet

INSERT INTOを実行して既存の外部テーブルにデータを挿入

構文

INSERT INTO external_schema.table_name
{ select_statement }

追加するテーブルの作成

最初に準備で作成したテーブルと同様の形式で、INSERT用テーブル（order_inc）を用意しました。このテーブルのデータはヘッダ付きのTSVファイルを出力します。最初はデータが入っていないので'numRows'='0'と作成しています。

cmdb=# create external table cm_user_db.order_inc(
cmdb(# row_id bigint,
cmdb(# order_id varchar(32),
cmdb(# order_date varchar(10),
cmdb(# ship_date varchar(10),
cmdb(# ship_mode varchar(64),
cmdb(# customer_id varchar(64),
cmdb(# customer_name varchar(64),
cmdb(# segment varchar(64),
cmdb(# country varchar(16),
cmdb(# city varchar(16),
cmdb(# state varchar(16),
cmdb(# region varchar(16),
cmdb(# product_id varchar(255),
cmdb(# category varchar(16),
cmdb(# sub_category varchar(32),
cmdb(# product_name varchar(255),
cmdb(# sales double precision,
cmdb(# quantity bigint,
cmdb(# discount double precision,
cmdb(# profit double precision)
cmdb-# row format delimited
cmdb-# fields terminated by '\t'
cmdb-# stored as textfile
cmdb-# location 's3://cm-user/orders_inc/'
cmdb-# table properties ('numRows'='0', 'skip.header.line.count'='1');
CREATE EXTERNAL TABLE

データの追加（１回目）

新規に作成したテーブルに2014年度のデータをINSERT INTO SELECTで追加しました。

cmdb=# insert into cm_user_db.order_inc (
cmdb(# select * from cm_user_db.order
cmdb(# where order_date >= '2014-01-01' and order_date <= '2014-12-31' );
INFO:  1680 record(s) exported successfully.
INSERT

結果の確認（１回目）

2014年度のデータのみが格納されています。

cmdb=# select min(order_date) as mindate, max(order_date) as maxdate from cm_user_db.order_inc;

  mindate   |  maxdate
------------+------------
 2014-01-03 | 2014-12-31
(1 row)

データファイルは以下のように作成されています。指定したS3のパスには４つに分割されたTSVファイルが出力されました。

% aws s3 ls s3://cm-user/orders_inc/
2020-06-09 09:03:19      95696 20200609_000314_797670_1486470_0000_part_00
2020-06-09 09:03:19     102849 20200609_000314_797670_1486470_0001_part_00
2020-06-09 09:03:19     103201 20200609_000314_797670_1486470_0002_part_00
2020-06-09 09:03:19     106458 20200609_000314_797670_1486470_0003_part_00

データの追加（２回目）

作成したテーブルに2015年度のデータをINSERT INTO SELECTで追加しました。

cmdb=# insert into cm_user_db.order_inc (
cmdb(# select * from cm_user_db.order
cmdb(# where order_date >= '2015-01-01' and order_date <= '2015-12-31') ;
INFO:  2389 record(s) exported successfully.
INSERT

結果の確認（２回目）

2014年度と2015年度のデータが格納されています。

cmdb=# select min(order_date) as mindate, max(order_date) as maxdate from cm_user_db.order_inc;
  mindate   |  maxdate
------------+------------
 2014-01-03 | 2015-12-31
(1 row)

データファイルはさらに４つのファイルが追加されていることが確認できました。つまり、INSERT INTO SELECTを繰り返すたびに追加したデータファイルが追加されるということです。

% aws s3 ls s3://cm-user/orders_inc/
2020-06-09 09:03:19      95696 20200609_000314_797670_1486470_0000_part_00
2020-06-09 09:03:19     102849 20200609_000314_797670_1486470_0001_part_00
2020-06-09 09:03:19     103201 20200609_000314_797670_1486470_0002_part_00
2020-06-09 09:03:19     106458 20200609_000314_797670_1486470_0003_part_00
2020-06-09 09:08:59     155395 20200609_000857_833399_1486552_0000_part_00
2020-06-09 09:08:59     138457 20200609_000857_833399_1486552_0001_part_00
2020-06-09 09:08:59     138305 20200609_000857_833399_1486552_0002_part_00
2020-06-09 09:08:59     147138 20200609_000857_833399_1486552_0003_part_00

データカタログの確認

Amazon AthenaからDDLを生成して、TBLPROPERTIESのnumRowsを確認したところ追加したレコードの件が更新されていることが確認されました。つまり、データの増加に伴いデータカタログのレコード数を更新しているということです。データカタログのレコード数が更新されることは、クエリエンジンのクエリプランナがこれらの情報を参照できることになります。

：
TBLPROPERTIES (
  'numRows'='4069', 
  'skip.header.line.count'='1', 
  'transient_lastDdlTime'='1591661339')

最後に

Amazon Athenaで提供済みの「クエリの結果から外部テーブルを作成する（CTAS）」と「追加するテーブルの作成」が、Redshift Spectrumでも利用できるようになりました。Amazon Athenaは主に静的なデータ（イミュータブルなデータ）を取り扱う事が多いですが、今後は、Amazon Redshiftが保持する常に変化する大量のデータを対象に外部テーブルとそのデータファイルを作成できるようになるので、ストレージとコンピューティングが分離されたRedshiftが、データレイクの中心的な役割を担えるようになりました。個人的な要望としては、外部テーブルに対するTRUNCATEがサポートされることを期待します。

合わせて読みたい